Advanced Operators

RxJava তে অনেক ধরনের অপারেটর রয়েছে যা ডেটার ফ্লো ও স্ট্রিমের ওপর শক্তিশালী নিয়ন্ত্রণ প্রদান করে। অ্যাডভান্সড অপারেটরগুলো সাধারণ অপারেটরগুলোর চেয়ে বেশি কাস্টমাইজেশন এবং ফ্লেক্সিবিলিটি প্রদান করে। এগুলো ডেটার প্রক্রিয়া, ফিল্টারিং, ট্রান্সফর্মেশন এবং স্ট্রিম সমন্বয়ের জন্য অত্যন্ত কার্যকরী।

অ্যাডভান্সড অপারেটরের প্রধান ধরণ

FlatMap

flatMap অপারেটরটি একটি ইনপুট আইটেমকে অনেকগুলো নতুন আইটেমে রূপান্তর করতে ব্যবহৃত হয়। এটি সাধারণত একাধিক স্ট্রিমকে একত্রিত করতে ব্যবহার করা হয়। flatMap ব্যবহার করে একটি স্ট্রিম থেকে অন্য স্ট্রিমে ডেটা পাঠানো সম্ভব হয়।

Observable<Integer> source = Observable.just(1, 2, 3, 4);
source
    .flatMap(i -> Observable.just(i * 10, i * 100))  // প্রতিটি মানকে দুটো ভিন্ন মানে রূপান্তর
    .subscribe(System.out::println);
Copy

এখানে flatMap প্রত্যেক ইনপুট মানের জন্য দুটি নতুন মান তৈরি করে এবং সেগুলোকে একটি একক স্ট্রিমে রূপান্তরিত করে।

ConcatMap

concatMap অপারেটরটি flatMap এর মতো, কিন্তু এটি স্ট্রিমগুলোর অর্ডার বজায় রাখে। এটি স্ট্রিমের ভ্যালুগুলিকে একটি নির্দিষ্ট সিকোয়েন্সে প্রসেস করতে ব্যবহৃত হয়।

Observable<Integer> source = Observable.just(1, 2, 3, 4);
source
    .concatMap(i -> Observable.just(i * 10, i * 100))  // একইভাবে কাজ করে, কিন্তু ফলাফল নির্দিষ্ট অর্ডারে আসে
    .subscribe(System.out::println);
Copy

এটি flatMap এর মতো স্ট্রিম প্রসেস করে, তবে এখানে স্ট্রিমের অর্ডার বজায় থাকে।

Zip

zip অপারেটরটি দুটি বা তার বেশি Observable কে একত্র করে এবং তাদের একেকটি আইটেম মেলানোর জন্য একটি ফাংশন ব্যবহার করে। এটি সাধারণত যখন একাধিক স্ট্রিম থেকে ডেটা একত্রিত করার প্রয়োজন হয়, তখন ব্যবহৃত হয়।

Observable<Integer> source1 = Observable.just(1, 2, 3);
Observable<Integer> source2 = Observable.just(10, 20, 30);
Observable.zip(source1, source2, (i1, i2) -> i1 + i2)
    .subscribe(System.out::println);
Copy

এখানে দুটি Observable এর মান একত্রিত করা হচ্ছে এবং তাদের যোগফল প্রিন্ট করা হচ্ছে। zip নিশ্চিত করে যে দুটি স্ট্রিম একসঙ্গে প্রসেস হয় এবং তাদের মান জোড়া আকারে মেলানো হয়।

Buffer

buffer অপারেটরটি একটি নির্দিষ্ট সংখ্যক আইটেম সংগ্রহ করে এবং পরে সেগুলোর একটি ব্যাচ রূপে প্রসেস করে। এটি একটি ব্যাচের ডেটাকে একত্রিত করতে ব্যবহৃত হয়, যাতে বৃহৎ ডেটার সাথে কাজ করতে সুবিধা হয়।

Observable<Integer> source = Observable.just(1, 2, 3, 4, 5, 6, 7, 8);
source
    .buffer(3)  // প্রতি ৩টি আইটেমে একটি ব্যাচ তৈরি করবে
    .subscribe(System.out::println);
Copy

এখানে প্রতি ৩টি আইটেমের একটি ব্যাচ তৈরি হচ্ছে এবং প্রতিটি ব্যাচ প্রিন্ট করা হচ্ছে।

GroupBy

groupBy অপারেটরটি ডেটাকে গ্রুপ করে নির্দিষ্ট শর্তের ভিত্তিতে। এটি খুবই কার্যকরী যখন আপনি ডেটাকে বিভিন্ন ক্যাটেগরিতে ভাগ করতে চান।

Observable<Integer> source = Observable.just(1, 2, 3, 4, 5, 6);
source
    .groupBy(i -> i % 2)  // সংখ্যাগুলোকে even এবং odd গ্রুপে ভাগ করবে
    .subscribe(group -> group.subscribe(System.out::println));
Copy

এখানে সংখ্যাগুলোকে even এবং odd হিসেবে গ্রুপ করা হচ্ছে এবং তাদের আলাদা করে প্রিন্ট করা হচ্ছে।

Debounce

debounce অপারেটরটি একটি সময় ফ্রেমে শুধুমাত্র সর্বশেষ ইভেন্টটিকে গ্রহণ করে, পুরোনো ইভেন্টগুলো বাতিল করে দেয়। এটি সাধারণত ইউজার ইন্টারঅ্যাকশনের ক্ষেত্রে ব্যবহৃত হয়, যেমন সার্চ বারে টাইপ করার সময়।

Observable<String> source = Observable.just("a", "b", "c", "d", "e");
source
    .debounce(500, TimeUnit.MILLISECONDS)  // ৫০০ মিলিসেকেন্ডের মধ্যে শুধুমাত্র শেষটি গ্রহণ করবে
    .subscribe(System.out::println);
Copy

এখানে debounce ব্যবহৃত হয়েছে যাতে দ্রুত ইভেন্টগুলোর মধ্যে শুধুমাত্র শেষ ইভেন্টটি প্রসেস হয়।

উপসংহার

RxJava তে অ্যাডভান্সড অপারেটরগুলি ডেটা স্ট্রিমের ওপর আরও সূক্ষ্ম নিয়ন্ত্রণ এবং কাস্টমাইজেশন প্রদান করে। এগুলোর মাধ্যমে একাধিক স্ট্রিমের সমন্বয়, ডেটা ট্রান্সফর্মেশন, এবং অর্ডার কন্ট্রোল সম্ভব হয়। এটি অ্যাসিঙ্ক্রোনাস ও রিঅ্যাকটিভ প্রোগ্রামিংয়ে আরও শক্তিশালী এবং উন্নত কাজের সুযোগ তৈরি করে।

RxJava এ SwitchMap(), GroupBy(), Window(), এবং Debounce() হল কিছু শক্তিশালী অপারেটর, যেগুলো রিঅ্যাকটিভ প্রোগ্রামিংয়ে বিশেষ ভূমিকা পালন করে। এই অপারেটরগুলো বিভিন্ন পরিস্থিতিতে ডেটার স্ট্রিমের উপর বিভিন্ন ধরনের প্রক্রিয়া বা পরিবর্তন করার সুবিধা দেয়। নিচে প্রতিটি অপারেটরের বিস্তারিত ব্যাখ্যা এবং ব্যবহার দেখানো হলো।

SwitchMap() অপারেটরের ব্যবহার

SwitchMap() একটি ট্রান্সফর্মেশন অপারেটর, যা মূলত একটি Observable থেকে প্রাপ্ত ভ্যালুকে নতুন একটি Observable এ ম্যাপ করে এবং পুরানো Observable এর সাবস্ক্রিপশনটি বাতিল করে দেয়। এর প্রধান সুবিধা হল, এটি ডেটার স্ট্রিমে দ্রুত পরিবর্তনগুলোর ক্ষেত্রে কাজ করে।

উদাহরণ:

ধরা যাক, আপনি একটি সার্চ ফিচার তৈরি করছেন যেখানে ইউজার যেকোনো সময় সার্চ টেক্সট পরিবর্তন করতে পারে। SwitchMap() অপারেটরটি ব্যবহার করলে, আগের সার্চ ফলাফলগুলি ইগনোর করা হবে এবং নতুন সার্চের জন্য নতুন সাবস্ক্রিপশন শুরু হবে।

Observable<String> searchObservable = searchEditText.getTextChanges();

searchObservable
    .switchMap(query -> searchApi.search(query))  // সার্চ API কল
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> {
        // নতুন সার্চ ফলাফল হ্যান্ডলিং
    });
Copy

এখানে, যখন ইউজার নতুন কিছু লিখবে, SwitchMap() আগের সার্চ ফলাফলকে বাতিল করে নতুন সার্চ স্ট্রিম শুরু করবে।

GroupBy() অপারেটরের ব্যবহার

GroupBy() অপারেটরটি একটি Observable এর উপাদানগুলোকে গ্রুপ করে দেয় একটি কী (key) এর ভিত্তিতে। এটি ব্যবহার করে আপনি একাধিক গ্রুপের জন্য পৃথক স্ট্রিম তৈরি করতে পারেন এবং প্রতিটি গ্রুপের উপর আলাদা কাজ করতে পারেন।

উদাহরণ:

ধরা যাক, আপনার কাছে কিছু ইউজারের ডেটা রয়েছে, এবং আপনি এই ডেটাকে বয়সের ভিত্তিতে গ্রুপ করতে চান:

Observable<User> usersObservable = getUsersObservable();

usersObservable
    .groupBy(user -> user.age)  // বয়সের ভিত্তিতে গ্রুপ করা
    .flatMap(group -> group.toList())  // প্রতিটি গ্রুপকে একটি তালিকায় রূপান্তর করা
    .subscribe(groupedUsers -> {
        // প্রতিটি বয়স গ্রুপের জন্য কাজ করা
    });
Copy

এখানে, groupBy() ইউজারদের বয়সের ভিত্তিতে গ্রুপ করে এবং প্রতিটি গ্রুপের জন্য আলাদা অপারেশন করতে দেয়।

Window() অপারেটরের ব্যবহার

Window() অপারেটরটি একটি স্ট্রিমকে একটি নির্দিষ্ট আকার বা সময়ের উইন্ডোতে বিভক্ত করে। এটি সাধারণত স্ট্রিমের সেগমেন্ট বা উইন্ডো তৈরি করতে ব্যবহৃত হয়, যা বড় স্ট্রিমগুলিকে ছোট অংশে ভাগ করে।

উদাহরণ:

ধরা যাক, আপনি একটি ইনফিনিট ডেটা স্ট্রিম থেকে প্রতি 5 সেকেন্ড পর পর একটি উইন্ডো তৈরি করতে চান:

Observable<Long> ticks = Observable.interval(1, TimeUnit.SECONDS);

ticks
    .window(5, TimeUnit.SECONDS)  // প্রতি 5 সেকেন্ড পর একটি উইন্ডো তৈরি
    .flatMap(window -> window.toList())  // প্রতিটি উইন্ডোকে একটি লিস্টে রূপান্তর করা
    .subscribe(windowedData -> {
        // উইন্ডোড ডেটা প্রক্রিয়া করা
    });
Copy

এখানে, window() প্রতি 5 সেকেন্ড পর একটি নতুন উইন্ডো তৈরি করবে, যেখানে প্রতি 5 সেকেন্ডে পাওয়া ডেটা একত্রিত হবে।

Debounce() অপারেটরের ব্যবহার

Debounce() অপারেটরটি সাধারণত ডেটার দ্রুত পরিবর্তন রোধ করতে ব্যবহৃত হয়, যেখানে আপনি নিশ্চিত করতে চান যে, শুধুমাত্র কিছু সময় পর ডেটা গ্রহণ করা হবে। এটি বিশেষত ইউজার ইনপুটের ক্ষেত্রে কার্যকর, যেমন টাইপ করার সময়।

উদাহরণ:

ধরা যাক, আপনি একটি সার্চ বক্স তৈরি করেছেন এবং ইউজার যত দ্রুত টাইপ করবে, সার্চ কোডটি বারবার না চলার জন্য debounce() ব্যবহার করতে পারেন:

Observable<String> searchObservable = searchEditText.getTextChanges();

searchObservable
    .debounce(300, TimeUnit.MILLISECONDS)  // 300 মিলিসেকেন্ড পর পর শুধুমাত্র একটি সার্চ কল হবে
    .switchMap(query -> searchApi.search(query))  // সার্চ API কল
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> {
        // সার্চ ফলাফল প্রক্রিয়া করা
    });
Copy

এখানে, debounce() ইউজারের টাইপিংয়ের মধ্যে 300 মিলিসেকেন্ডের জন্য বিলম্ব করবে, যাতে করে টাইপিং শেষ হওয়ার পরে শুধুমাত্র একটি সার্চ কল হবে।

উপসংহার

SwitchMap(), GroupBy(), Window(), এবং Debounce() হল RxJava এর কিছু শক্তিশালী অপারেটর যা ডেটার স্ট্রিমে বিভিন্ন প্রকারের ট্রান্সফরমেশন এবং ফিল্টারিংয়ের কাজ সহজ করে তোলে। এই অপারেটরগুলো ব্যবহার করে অ্যাসিঙ্ক্রোনাস ডেটা স্ট্রিম প্রক্রিয়া এবং রিঅ্যাকটিভ প্রোগ্রামিং আরও কার্যকরী এবং স্কেলেবল হতে পারে। এগুলো বড় ডেটাবেস বা API কলের জন্য পারফরম্যান্স উন্নত করতে সাহায্য করে এবং ইউজার ইন্টারফেসে প্রাকটিক্যাল কাজগুলো সহজ করে।

RxJava তে time-based এবং conditional operators ব্যবহার করে আপনি স্ট্রিমের উপর নির্দিষ্ট সময়ের ভিত্তিতে অপারেশন করতে পারেন এবং শর্ত (conditions) অনুসারে ডেটা প্রক্রিয়া করতে পারেন। এই অপারেশনগুলো অ্যাসিঙ্ক্রোনাস প্রোগ্রামিংয়ের ক্ষেত্রে আরও ফ্লেক্সিবল এবং কার্যকরী হতে সাহায্য করে। বিশেষত, যখন আপনি নির্দিষ্ট সময়ের মধ্যে ডেটা ইভেন্টগুলোর উপর কাজ করতে চান বা নির্দিষ্ট শর্ত অনুযায়ী কোনো অ্যাকশন ট্রিগার করতে চান, তখন এই অপারেশনগুলো অত্যন্ত উপকারী।

Time-based Operators

Time-based operators RxJava তে আপনাকে নির্দিষ্ট সময়ের ভিত্তিতে ডেটা প্রসেসিং করার সুবিধা দেয়। এটি সময়-ভিত্তিক স্ট্রিম তৈরি করতে ব্যবহার করা হয়, যেমন সময় নির্ধারণ করে সাবস্ক্রিপশন শুরু বা থামানো। কিছু প্রচলিত time-based operators হলো:

Delay

delay অপারেশনটি নির্দিষ্ট সময় (milliseconds বা seconds) পর্যন্ত একটি স্ট্রিমের আইটেম প্রক্রিয়া করার জন্য ব্যবহৃত হয়। এটি স্ট্রিমে আইটেম আসার আগে বিলম্ব (delay) যোগ করে।

Observable.just("Hello", "World")
    .delay(2, TimeUnit.SECONDS) // 2 সেকেন্ড দেরি
    .subscribe(System.out::println);
Copy

এখানে delay অপারেশনটি 2 সেকেন্ড পর "Hello" এবং "World" প্রিন্ট করবে।

Timer

timer অপারেশনটি একটি নির্দিষ্ট সময়ের পরে একটি একক আইটেম (যেমন 0) ইমিট করবে। এটি সাধারনত একক ইভেন্ট ট্রিগার করার জন্য ব্যবহৃত হয়।

Observable.timer(3, TimeUnit.SECONDS)
    .subscribe(time -> System.out.println("Time's up!"));
Copy

এটি 3 সেকেন্ড পর "Time's up!" প্রিন্ট করবে।

Interval

interval অপারেশনটি একটি নির্দিষ্ট সময় ব্যবধানে ধারাবাহিকভাবে ডেটা ইভেন্ট উৎপন্ন করে। এটি নির্দিষ্ট সময়ের পর পর (যেমন প্রতি 1 সেকেন্ড) একটি আইটেম পাঠায়।

Observable.interval(1, TimeUnit.SECONDS)
    .take(5) // 5টি ইভেন্ট নেবে
    .subscribe(System.out::println);
Copy

এটি প্রতি সেকেন্ডে একটি সংখ্যা ইমিট করবে এবং 5টি ইভেন্টের পর সাবস্ক্রিপশন বন্ধ হয়ে যাবে।

Timer vs Interval

• Timer: একটি একক ইভেন্টের জন্য সময় নির্ধারণ করে।
• Interval: ধারাবাহিকভাবে সময় নির্ধারণ করে এবং অনেক ইভেন্ট তৈরি করে।

Conditional Operators

Conditional operators ব্যবহার করে আপনি রিঅ্যাকটিভ স্ট্রিমে বিভিন্ন শর্ত অনুযায়ী অপারেশন পরিচালনা করতে পারেন। এই অপারেশনগুলো স্ট্রিমের মধ্যে শর্ত (condition) চেক করতে সাহায্য করে এবং শুধুমাত্র সেগুলোই প্রসেস করে যেগুলো শর্ত পূর্ণ করবে।

Take

take অপারেশনটি একটি স্ট্রিম থেকে প্রথম N সংখ্যক আইটেম নেয়। এটি শর্ত হিসেবে একটি সংখ্যা ব্যবহার করে, যা জানায় কতগুলো আইটেম প্রয়োজন।

Observable.just(1, 2, 3, 4, 5)
    .take(3) // প্রথম 3টি আইটেম নিবে
    .subscribe(System.out::println);
Copy

এটি শুধুমাত্র 1, 2, এবং 3 প্রিন্ট করবে।

TakeWhile

takeWhile অপারেশনটি একটি শর্ত অনুসারে স্ট্রিম থেকে আইটেম নেবে যতক্ষণ না শর্তটি মিথ্যা হয়। এটি শর্ত পরীক্ষা করে এবং যখন শর্ত মিথ্যা হবে, তখন স্ট্রিমটি বন্ধ হয়ে যায়।

Observable.just(1, 2, 3, 4, 5)
    .takeWhile(i -> i < 4) // যতক্ষণ না 4 এর চেয়ে ছোট, ততক্ষণ আইটেম নেবে
    .subscribe(System.out::println);
Copy

এটি 1, 2, এবং 3 প্রিন্ট করবে, কারণ 4 এর পর আইটেমের মান শর্ত পূর্ণ করবে না।

Skip

skip অপারেশনটি একটি স্ট্রিম থেকে প্রথম N সংখ্যক আইটেম বাদ দিয়ে বাকি আইটেমগুলো নেয়। এটি শর্ত হিসেবে একটি সংখ্যা ব্যবহার করে, যা জানায় কতগুলো আইটেম বাদ দিতে হবে।

Observable.just(1, 2, 3, 4, 5)
    .skip(2) // প্রথম 2টি আইটেম বাদ দিয়ে বাকি আইটেম নেবে
    .subscribe(System.out::println);
Copy

এটি 3, 4, এবং 5 প্রিন্ট করবে, কারণ প্রথম দুটি আইটেম (1 এবং 2) বাদ দেওয়া হয়েছে।

Filter

filter অপারেশনটি একটি শর্তের মাধ্যমে স্ট্রিমে থাকা আইটেমগুলো ফিল্টার করে এবং শুধুমাত্র সেগুলো নেয় যেগুলো শর্ত পূর্ণ করে। এটি শর্ত হিসেবে একটি predicate ব্যবহার করে।

Observable.just(1, 2, 3, 4, 5)
    .filter(i -> i % 2 == 0) // শুধু Even নম্বর নেবে
    .subscribe(System.out::println);
Copy

এটি 2 এবং 4 প্রিন্ট করবে, কারণ এই দুটি নম্বর Even (যুগল)।

উপসংহার

RxJava তে time-based এবং conditional operators ব্যবহারের মাধ্যমে আপনি ডেটার প্রক্রিয়া আরও নির্ভুল এবং কার্যকরী করতে পারেন। time-based operators যেমন delay, interval, এবং timer সময়ভিত্তিক স্ট্রিমের জন্য কার্যকর, যেখানে conditional operators যেমন take, filter, এবং takeWhile ডেটার নির্দিষ্ট শর্ত অনুযায়ী ফিল্টারিং এবং সিলেকশন করতে সাহায্য করে। এই অপারেশনগুলো আপনার অ্যাসিঙ্ক্রোনাস বা রিঅ্যাকটিভ প্রোগ্রামিংয়ের দক্ষতা বাড়াতে সহায়ক হবে।

RxJava তে, Advanced Operators বা উন্নত অপারেশনগুলো ব্যবহার করে আপনি আরও জটিল এবং উন্নত কার্যপ্রণালী তৈরি করতে পারেন। এগুলো সাধারণত ডেটা স্ট্রিমের সাথে আরও পুঙ্খানুপুঙ্খভাবে কাজ করতে এবং বিভিন্ন শর্তের ভিত্তিতে ডেটা প্রসেসিং করতে ব্যবহৃত হয়। এখানে কিছু গুরুত্বপূর্ণ উন্নত অপারেশন ও তাদের উদাহরণ দেয়া হল।

FlatMap Operator

flatMap() অপারেশনটি এক Observable থেকে আউটপুট পাওয়া মানগুলোর ওপর আরও একটি Observable তৈরি করতে ব্যবহৃত হয় এবং তারপর তাদের ফলাফলকে একত্রিত করে। এটি একাধিক স্ট্রিমের মাধ্যমে ডেটা প্রসেসিং করতে সাহায্য করে।

উদাহরণ:

Observable<Integer> observable = Observable.just(1, 2, 3);

observable.flatMap(num -> Observable.just(num * 2, num * 3))
    .subscribe(System.out::println);
Copy

এখানে, flatMap() অপারেশনটি প্রতিটি মানের জন্য নতুন দুটি স্ট্রিম তৈরি করছে, একটিতে মানটি ২ দিয়ে গুণিত হচ্ছে এবং অন্যটিতে ৩ দিয়ে গুণিত হচ্ছে। আউটপুট হবে:

2
3
4
6
6
9
Copy

SwitchMap Operator

switchMap() অপারেশনটি এমন একটি অপারেশন যা প্রথম Observable থেকে ডেটা পাওয়া সাপেক্ষে পরবর্তী Observable নির্বাচন করে। এটি সাধারণত তখন ব্যবহৃত হয় যখন আপনি নতুন কোনো স্ট্রিমে সুইচ করতে চান এবং পূর্ববর্তী স্ট্রিমের সাবস্ক্রিপশনটি ডিসপোজ করতে চান।

উদাহরণ:

Observable<Integer> observable = Observable.just(1, 2, 3);

observable.switchMap(num -> {
    if (num == 2) {
        return Observable.just(100, 200);
    }
    return Observable.just(10, 20);
})
.subscribe(System.out::println);
Copy

এখানে, যখন মান ২ হয়, তখন switchMap() একটি নতুন স্ট্রিম (100, 200) প্রেরণ করে। আউটপুট হবে:

10
20
100
200
Copy

MergeWith Operator

mergeWith() অপারেশনটি একাধিক Observable এর স্ট্রিম একত্রিত করে একটি একক স্ট্রিম তৈরি করতে ব্যবহৃত হয়। এটি merge() অপারেশনের মতো কাজ করে, তবে এটি একটি একক Observable উপর কাজ করে।

উদাহরণ:

Observable<Integer> observable1 = Observable.just(1, 2, 3);
Observable<Integer> observable2 = Observable.just(4, 5, 6);

observable1.mergeWith(observable2)
    .subscribe(System.out::println);
Copy

এখানে, observable1 এবং observable2 এর ডেটাগুলো একত্রিত হয়ে প্রদর্শিত হবে। আউটপুট হবে:

1
2
3
4
5
6
Copy

GroupBy Operator

groupBy() অপারেশনটি একটি Observable এর মানগুলোকে নির্দিষ্ট কriteria এর উপর ভিত্তি করে গ্রুপ করে এবং প্রতিটি গ্রুপের জন্য আলাদা Observable তৈরি করে। এটি মূলত ডেটাকে শ্রেণীবদ্ধ করতে ব্যবহৃত হয়।

উদাহরণ:

Observable<Integer> observable = Observable.just(1, 2, 3, 4, 5, 6);

observable.groupBy(num -> num % 2 == 0 ? "Even" : "Odd")
    .subscribe(group -> group.subscribe(value -> 
        System.out.println(group.getKey() + ": " + value)
    ));
Copy

এখানে, groupBy() অপারেশনটি সংখ্যাগুলোকে দুটি গ্রুপে ভাগ করবে: Even এবং Odd। আউটপুট হবে:

Odd: 1
Odd: 3
Odd: 5
Even: 2
Even: 4
Even: 6
Copy

CombineLatest Operator

combineLatest() অপারেশনটি একাধিক Observable এর সর্বশেষ মানগুলিকে একত্রিত করে এবং একটি নতুন মান তৈরি করে। এটি মূলত তখন ব্যবহৃত হয় যখন একাধিক স্ট্রিমের সর্বশেষ ডেটা থেকে একটি নতুন ফলাফল তৈরি করতে হয়।

উদাহরণ:

Observable<Integer> observable1 = Observable.just(1, 2, 3);
Observable<String> observable2 = Observable.just("A", "B");

Observable.combineLatest(observable1, observable2, (num, letter) -> num + letter)
    .subscribe(System.out::println);
Copy

এখানে, combineLatest() অপারেশনটি observable1 এবং observable2 এর সর্বশেষ মান থেকে একটি নতুন মান তৈরি করবে। আউটপুট হবে:

3B
Copy

ConcatMap Operator

concatMap() অপারেশনটি একটি Observable থেকে আউটপুট পাওয়ার পর পরবর্তী Observable তৈরি করতে ব্যবহৃত হয়। এটি flatMap() এর মতো, তবে এক্ষেত্রে সব সাবস্ক্রিপশন একটি নির্দিষ্ট ক্রমে সম্পন্ন হয়, অর্থাৎ প্রথম স্ট্রিমের ডেটা সম্পূর্ণ না হওয়া পর্যন্ত পরবর্তী স্ট্রিম শুরু হয় না।

উদাহরণ:

Observable<Integer> observable = Observable.just(1, 2, 3);

observable.concatMap(num -> Observable.just(num * 2, num * 3))
    .subscribe(System.out::println);
Copy

এখানে, concatMap() অপারেশনটি প্রথমে ১ এর জন্য সব ডেটা প্রেরণ করবে, তারপর ২ এবং ৩ এর জন্য। আউটপুট হবে:

2
3
4
6
6
9
Copy

উপসংহার

RxJava তে Advanced Operators এর মাধ্যমে আপনি আরও জটিল এবং শক্তিশালী অ্যাসিঙ্ক্রোনাস কাজ করতে পারেন। flatMap(), switchMap(), groupBy(), mergeWith() এবং অন্যান্য অপারেশনগুলো ডেটা স্ট্রিমের সাথে গভীরভাবে কাজ করার জন্য কার্যকরী এবং নমনীয় পদ্ধতি প্রদান করে। এই অপারেশনগুলো ব্যবহার করলে আপনি আরও দক্ষ এবং মানসম্পন্ন রিয়্যাকটিভ অ্যাপ্লিকেশন তৈরি করতে পারবেন।